package TopN;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.Iterator;
import java.util.Set;
import java.util.TreeMap;

public class topnReducer extends Reducer<Text,FlowBean,Text,FlowBean> {
    private TreeMap<FlowBean,Text> treeMap=new TreeMap<>();

    @Override
    protected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException {
        for(FlowBean value:values){
            //treeMap里面插入的必须是实例化的对象，现在的value只是一个引用变量（每一次遍历指向集合中的一个新的对象）并不是一个实例化的对象。
            FlowBean flowBean=new FlowBean();
            flowBean.set(value.getUpFlow(),value.getDownFlow());

            //此时的key也不是实例化对象，所以要先实例化一个对象。
            Text phone=new Text(key);
            treeMap.put(flowBean,phone);
        }

        if(treeMap.size()>10){
            treeMap.remove(treeMap.lastKey());
        }
    }

    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        Set<FlowBean> flowBeans = treeMap.keySet();
        Iterator<FlowBean> iterator = flowBeans.iterator();
        while(iterator.hasNext()){
            FlowBean flowBean=iterator.next();
            Text text=treeMap.get(flowBean);
            context.write(text,flowBean);
        }
    }
}
